package olegroshka.src.com.bofa.olegroshka.datasource;

import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;

import olegroshka.src.com.bofa.olegroshka.common.PrimitiveLatch;

/**
 * Joins two records (semantically similar to the left join in SQL).
 * 
 * @author Oleg
 * 
 */
public class ParallelJoinDataSource implements DataSource<Object[]> {
	
	private final static Logger LOG = Logger
			.getLogger(ParallelJoinDataSource.class.getName());

	private final DataSource<Object[]> leftDataSource;
	private final DataSource<Object[]> rightDataSource;
	
	private final int leftColumn;
	private final int rightColumn;
	private final int leftRecordLength;
	private final int rightRecordLength;
	private final int joinedRecordLength;
	
	private PrimitiveLatch rightSideReadySignal = new PrimitiveLatch(1);

 	private final Map<Object, Object[]> rightMap = 
			Collections.synchronizedMap(new LinkedHashMap<Object, Object[]>(1000)); 
	
	public ParallelJoinDataSource(DataSource<Object[]> leftDataSource,
			DataSource<Object[]> rightDataSource, int leftColumn,
			int rightColumn, int leftRecordSize, int rightRecordSize) {
		this.leftDataSource = leftDataSource;
		this.rightDataSource = rightDataSource;
		this.leftColumn = leftColumn;
		this.rightColumn = rightColumn;
		this.leftRecordLength = leftRecordSize;
		this.rightRecordLength = rightRecordSize;
		this.joinedRecordLength = leftRecordSize + rightRecordSize;
	}

	public Iterator<Object[]> iterator() {

		return new Iterator<Object[]>() {
			
			private final Iterator<Object[]> iterator = 
					leftDataSource.iterator();
			
			private Object[] currentRecord = null;

			public boolean hasNext() {
				try {
					rightSideReadySignal.await();
				}
				catch (InterruptedException e) {
					LOG.info("Interrupted by user.");
					Thread.currentThread().interrupt();
				}
				if (currentRecord == null) {
					currentRecord = next();
				}
				return currentRecord != null;
			}

			public Object[] next() {
				Object[] joinedRecord = null;
				try {
					rightSideReadySignal.await();
					
					joinedRecord = currentRecord;
					if (joinedRecord != null || !iterator.hasNext()) {
						currentRecord = null;
						return joinedRecord;
					}
					Object[] leftRecord = 
							(Object[]) iterator.next();
					Object key = leftRecord[leftColumn];
					Object[] rightRecord = rightMap.get(key);
					if( rightRecord != null ) {
						joinedRecord = new Object[joinedRecordLength];   
						System.arraycopy(leftRecord, 0, joinedRecord, 0, leftRecordLength);
						System.arraycopy(rightRecord, 0, joinedRecord, leftRecordLength, rightRecordLength);						
						rightMap.remove(key);
					} else {
						joinedRecord = new Object[joinedRecordLength];   
						System.arraycopy(leftRecord, 0, joinedRecord, 0, leftRecordLength);						
					}
				}
				catch (InterruptedException e) {
					LOG.info("Interrupted by user.");
					Thread.currentThread().interrupt();
				}
				
				return joinedRecord;
			}

			public void remove() {
				throw new UnsupportedOperationException();
			}
		};
	}

	public void begin() {
		rightSideReadySignal = new PrimitiveLatch(1);
		leftDataSource.begin();
		rightDataSource.begin();
		indexRightRecords();
	}

	public void indexRightRecords() {
		//store in the hash map right side   
		new Thread(new Runnable() {
			public void run() {
				Iterator<Object[]> iterator = rightDataSource.iterator();
				while (iterator.hasNext()) {
					Object[] record = (Object[]) iterator.next();
					Object key = record[rightColumn];
					if( key != null ) {
						rightMap.put(key, record);
					} else {
						LOG.log(Level.WARNING, "Ignoring record: " + record + 
								", key value is null, column=" + leftColumn);
					}
				}
				rightSideReadySignal.countDown();
			}
		}).start();
	}

	public void end() {
		rightMap.clear();
		leftDataSource.end();
		rightDataSource.end();
	}
}
